from multiprocessing import Pool, Manager
from lxml import etree
from queue import Queue
from requests.auth import HTTPProxyAuth
from tool import *
from sql import sqlseverDB
import requests, concurrent.futures
import traceback,logging,math

ms = sqlseverDB('{SQL Server}', SQL_SERVER, 'SuMaiTongPol', SQL_USER, SQL_PWD)

mutil_count=5

logger = logging.getLogger()
logger.setLevel(logging.INFO)  # 设置日志级别
log_path=f'日志文件/亚马逊CJSeller.log'
file_handler = logging.FileHandler(log_path, mode='a', encoding='utf-8')

# 创建控制台处理器 (Handler)
console_handler = logging.StreamHandler()
console_handler.setLevel(logging.INFO)

# 设置日志格式
formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')
file_handler.setFormatter(formatter)
console_handler.setFormatter(formatter)

# 将处理器添加到 logger 对象
logger.addHandler(file_handler)
logger.addHandler(console_handler)

def process_shop(bid,kitems,queue,task_count):
    logging.info(f"进程启动: bid={bid}, 任务数={len(kitems)}")
    try:
        res= asyncio.run(reqPro(bid,kitems,queue,task_count))
        logging.info(res)

    except Exception as e:
        logging.error(f'bid:{bid}, 出现错误: {e}')
        traceback.print_exc()
        time.sleep(3)

async def reqPro(bid,pitem,ycj_queue,task_count):

    browser=await openSunBrowser(bid)
    page=await browser.newPage()
    for pid, asin, purl in pitem:
        x_ind=ycj_queue.qsize()+1
        try:
            await page.goto(purl,{'timeout':0})
            await asyncio.sleep(3)
        except Exception as e:
            print(f'第{x_ind}个,ASIN:{asin},,请求出错 => {e}')
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            continue
        elSeller=None
        while True:
            cot=await page.content()
            el_tit=await page.xpath('//span[@id="productTitle"]')
            if el_tit:
                elSeller = etree.HTML(cot).xpath('//a[contains(@href,"/gp/help/seller")]/@href')
                if elSeller:
                    break
                else:
                    aff_no_seller=ms.ExecNoQuery('update [AmazonAsin] set state=2 where [ASIN]=?', (asin,))
                    print(f'第{x_ind}个,ASIN:{asin},,产品页找不到卖家地址,内容长度:{len(cot)},标记状态2:{aff_no_seller}')
                    break
            
            cur_tit= await page.title()
            if cur_tit=='Sorry! Something went wrong!':
                aff_503=ms.ExecNoQuery('update [AmazonAsin] set state=503 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,ASIN:{asin},页面503,标记状态503:{aff_503}')
                break
            if 'Page Not Found' in cur_tit:
                aff_404=ms.ExecNoQuery('update [AmazonAsin] set state=404 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,ASIN:{asin},页面404,标记状态2:{aff_404}')
            cot= await page.content()
            if len(cot)<20000:
                aff_lower=ms.ExecNoQuery('update [AmazonAsin] set state=101 where [ASIN]=?', (asin,))
                logging.info(f'第{x_ind}个,ASIN:{asin},页面内容过少,标记状态101:{aff_lower}')
                break
        
        if not elSeller:
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            continue
        
        sellerUlr = f'https://www.amazon.com{elSeller[0]}'
        try:
            await page.goto(sellerUlr,{'timeout':0})
            await asyncio.sleep(3)
            cot2=await page.content()

        except Exception as e:
            print(f'第{x_ind}个,ASIN:{asin},,请求店铺网址出错 => {e}')
            traceback.print_exc()
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            continue
        
        sss=await wait_one_xpath(page,'//h1[@id="seller-name"]',max_wait=60)
        if not sss:
            print(f'第{x_ind}个,ASIN:{asin},,等待地址加载超时,跳过....')
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            continue

        el_sellerdes = etree.HTML(cot2).xpath('//div[@id="seller-desc-column"]//a[contains(@href,"marketplaceID")]/@href')
        el_loction = etree.HTML(cot2).xpath('//div[@id="page-section-detail-seller-info"]/div[1]/div[1]/div[1]/div[last()]//text()')
        if el_sellerdes and el_loction:
            dpurl = f'https://www.amazon.com{el_sellerdes[0]}'
            sellerLocation = el_loction[0].strip().lower()
            mid = get_url_param(dpurl, 'me')
            print(f'第{x_ind}个,ASIN:{asin},成功采集卖家内容 => mid:{mid},卖家地址:{sellerLocation}')

        else:
            aff_nf=ms.ExecNoQuery('update [AmazonAsin] set state=-2 where [ASIN]=?', (asin,))
            ycj_queue.put(1)
            latest_count=task_count-ycj_queue.qsize()
            print(f'第{x_ind}个,ASIN:{asin},,卖家页面找不到店铺首页网址,更改状态-2:{aff_nf}')
            continue

        if sellerLocation == 'cn':
            rsjc = ms.ExecQuery('select count(*) from AmazonSeller where mid=?', (mid,))
            if rsjc[0][0] == 0:
                aff1 = ms.ExecNoQuery('insert into AmazonSeller(MID, fromAsin, [TzURl], [Dpurl]) values (?,?,?,?)', (mid, asin, sellerUlr, dpurl))
            else:
                aff1 = '已存在店铺'
        else:
            aff1='非中国店铺'
        aff2 = ms.ExecNoQuery('update [AmazonAsin] set state=1,isuse=isuse+1 where [ASIN]=?', (asin,))
        
        ycj_queue.put(1)
        latest_count=task_count-ycj_queue.qsize()
        print(f'第{x_ind}个,ASIN:{asin},,采集状态:{aff1},更改状态:{aff2},剩余任务队列:{latest_count}/{task_count}')

    await page.close()
def main():
    
    print(f'第二步采集中国卖家店铺首页....')
    while True:
        rs_isuse=ms.ExecQuery('select min(isuse) from AmazonAsin') #先获取采集最小次数
        min_use_count=rs_isuse[0][0]
        rs = ms.ExecQuery(f'select [id], [ASIN], [Purl] from AmazonAsin where state=0 and isuse={min_use_count}')
        if len(rs)>0:
            break
        else:
            aff=ms.ExecNoQuery('update AmazonAsin set state=0 where state=1')
            print(f'ASIN最小采集次数:{min_use_count},已经全部采集完毕,将所有 {aff} 个ASIN采集状态置0')

    taskCount = len(rs)

    print(f'本次采集,ASIN最小使用次数:{min_use_count},待采集ASIN总数: {taskCount}')

    #创建采集环境
    while True:
        all_bids=get_browser_environment()
        print('获取到',all_bids)
        if len(all_bids)>=mutil_count:
            bids=all_bids[:mutil_count]
            break
        else:
            print(mutil_count-len(all_bids))
            for _ in range(mutil_count-len(all_bids)):
                res=create_browser_environment_no_proxy(f'亚马逊采集{get_random_code()}')
                time.sleep(1)
    
    print(f'成功创建:{len(bids)}个采集环境')

    manager = Manager()
    queue = manager.Queue()

    qu_jian=math.ceil(taskCount/mutil_count)
    with Pool(processes=mutil_count) as pool:  # 根据实际情况设置并发进程数量
        results = []
        for j in range(mutil_count):
            pitems=rs[j*qu_jian:(j+1)*qu_jian]
            result = pool.apply_async(
                process_shop, args=(bids[j],pitems,queue,taskCount))
            results.append(result)

        # 监控任务执行情况
        pool.close()
        for result in results:
            result.wait()  # 等待每个进程完成，无超时限制

        # 所有任务执行完成后继续
        pool.join()


if __name__ == '__main__':
    main()
